﻿// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using DotNetFramework.CAP.Core.Logger;
using DotNetFramework.CAP.Internal;
using DotNetFramework.CAP.Processor.States;
using Newtonsoft.Json;

namespace DotNetFramework.CAP.Mqtt
{
    public class MqttTransport : BasePublishMessageSender
    {
        private readonly IConnectionPool _connectionPool;
        private readonly ILogger _logger;

        public MqttTransport(ILogger logger, CapOptions options,
            IStorageConnection connection, IConnectionPool connectionChannelPool, IStateChanger stateChanger)
            : base(logger, options, connection, stateChanger)
        {
            _logger = logger;
            _connectionPool = connectionChannelPool;
            ServersAddress = connectionChannelPool.ServersAddress;
        }

        //public BrokerAddress BrokerAddress => new BrokerAddress("Kafka", _connectionPool.ServersAddress);

        public override async Task<OperateResult> PublishAsync(string keyName, string content)
        {
            var producer = _connectionPool.RentProducer();

            try
            {
                var mqMssage = new MqttMessage();

                await producer.PublishAsync(new MQTTnet.MqttApplicationMessage()
                {
                    Topic = keyName,
                    Payload = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(mqMssage))
                });

                //if (result.ReasonCode == MQTTnet.Client.Publishing.MqttClientPublishReasonCode.Success)
                //{
                //    _logger.LogDebug($"mqtt topic message [{message.GetName()}] has been published.");

                //    return Task.FromResult(OperateResult.Success);
                //}

                //throw new PublisherSentFailedException("mqtt message persisted failed!");
                return OperateResult.Success;
            }
            catch (Exception ex)
            {
                var wapperEx = new PublisherSentFailedException(ex.Message, ex);

                return OperateResult.Failed(wapperEx);
            }
            finally
            {
                _connectionPool.Return(producer);
            }
        }
    }
}